package Thread.mq;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.EventHandlerGroup;
import com.lmax.disruptor.dsl.ProducerType;
import com.lmax.disruptor.util.DaemonThreadFactory;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.zip.Inflater;

import static Thread.mq.Common.*;

/**
 * Created by sunyuming on 18/10/23.
 */
public class MyDisruptor {

    public static void main(String[] args) {

        CountDownLatch start = new CountDownLatch(1);
        CountDownLatch end = new CountDownLatch(1);

        DisruptorFactory disruptorFactory = new DisruptorFactory();
        ExecutorService executor = Executors.newFixedThreadPool(threadCount);
        Disruptor<MyInteger> disruptor = new Disruptor<MyInteger>(disruptorFactory, 1024*1024, executor,
                ProducerType.SINGLE, new YieldingWaitStrategy());
        DisruptorConsumer [] dis = new DisruptorConsumer[threadCount];
        for(int s = 0; s<threadCount; ++s) {
            dis[s] = new DisruptorConsumer(end);
        }
        disruptor.handleEventsWithWorkerPool(dis);
        disruptor.start();

        Runnable producer = new Producer(start, disruptor.getRingBuffer());

        for(int i=0; i<threadCount; ++i) {
            new Thread(producer).start();
        }

        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        start.countDown();
        try {
            end.await();
            System.out.println("耗时" + String.valueOf(ed-st));

        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private static class DisruptorConsumer implements WorkHandler<MyInteger> {

        private CountDownLatch countDownLatch;
        private volatile AtomicInteger ii = new AtomicInteger(0);
        public DisruptorConsumer(CountDownLatch countDownLatch) {
            this.countDownLatch = countDownLatch;
        }

        @Override
        public void onEvent(MyInteger myInteger) throws Exception {
            if(myInteger.last()) {
                ed = System.currentTimeMillis();
                countDownLatch.countDown();
                System.out.println(myInteger.mem);
            }
        }
    }

    private static class Producer implements Runnable {

        private RingBuffer<MyInteger> ringBuffer;
        private AtomicInteger atomicInteger = new AtomicInteger(0);

        CountDownLatch countDownLatch;
        public Producer(CountDownLatch countDownLatch, RingBuffer<MyInteger> ringBuffer) {
            this.countDownLatch = countDownLatch;
            this.ringBuffer = ringBuffer;
        }

        @Override
        public void run() {
            try {
                countDownLatch.await();
                st = System.currentTimeMillis();

                for(int i=0; i<LEN; ++i) {
                    long seq = ringBuffer.next();
                    MyInteger event = ringBuffer.get(seq);
                    event.mem = atomicInteger.incrementAndGet();
                    ringBuffer.publish(seq);
                }

            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    private static class DisruptorFactory implements EventFactory<MyInteger> {
        public MyInteger newInstance() {
            return new MyInteger(0);
        }
    }

}
